package com.behavioranalysis.flinkprogram.flink.ad;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple6;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * <h3>flinkprogram</h3>
 * <p>${description}</p>
 * Created by yang on 20-2-14 下午5:04
 * updated by yang on 20-2-14 下午5:04
 */
public class SourceString2TupleMapFunction implements
        MapFunction<String, Tuple6<Long, String, String, String, String, String>> {
    private static final long serialVersionUID = 1L;
    @Override
    public Tuple6<Long, String, String, String, String, String> map(String value) throws Exception {
        String[] splitedLogs = value.split(" ");
        long timepstamp = Long.valueOf(splitedLogs[0]);
        String province = splitedLogs[1];
        String city = splitedLogs[2];
        String userid = splitedLogs[3];
        String adid = splitedLogs[4];
        // 提取日期(yyyyMMdd)、userid、adid
        Date date = new Date(timepstamp);
        String datekey = new SimpleDateFormat("yyyyMMdd").format(date);

        // 拼接key
        String key = datekey + "_" + userid + "_" + adid;

        return new Tuple6<>(timepstamp, province, city, userid, adid, key);
    }
}
